package com.ls.mq.queue;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Future;

import org.apache.log4j.Logger;

import com.ls.mq.iterface.MessageListener;
import com.ls.mq.iterface.MessageService;
import com.ls.mq.redis.RedisFactory;
import com.ls.mq.thread.ExecutorsManager;

/**
 * 可以向channel发送和接收消息，并且把接收的消息交给对应的Listener来处理
 * 此代理一端监听redis，可以添加和删除对应的channel，channel的类型是list， 另一端提供Listener的注册和取消，注册特定channel的listener
 * 
 * @author Administrator
 *
 */
public class QueueProxy implements MessageService{
	Logger logger = Logger.getLogger(QueueProxy.class);
	private volatile boolean  isStarted=false;
	private volatile static QueueProxy instance;
	
	public static  QueueProxy getInstance(){
		if(instance==null)
			synchronized(QueueProxy.class){
				if(instance==null)
					instance=new QueueProxy();
			}
		return instance;
	}
	/**
	 * listener监听channel列表
	 */
	private Map<String, MessageListener> listenerMap;
	private Future<?> listenerFuture;
	private QueueProxy(){
		listenerMap= new HashMap<>();
	}
	
	public void addListener(String channel, MessageListener listener) {
		if(!isStarted)
			start();
		listenerMap.put(channel, listener);
	}
	
	public void removeListener(String channel){
		listenerMap.remove(channel);
	}

	public void publish(String channel, String msg) {
		RedisFactory.getInstance().getClient().lpush(channel, msg);
	}
	
	public synchronized void start(){
		if(isStarted) return;
		listenerFuture = ExecutorsManager.getInstance().submit(new ListenQueueMessageTask(getListenerMap()));
		isStarted=true;
	}

	public synchronized void stop(){
		if(!isStarted) return;
		if(listenerFuture!=null){
			boolean result = listenerFuture.cancel(true);
			logger.info("canel result "+result);
		}
		isStarted=false;
	}
	
	public Map<String, MessageListener> getListenerMap() {
		return listenerMap;
	}

}
